package main.scala
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Graph, _}
import org.apache.spark.graphx.lib.SVDPlusPlus
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, DataFrameReader, SparkSession}

object SvdBaseDemo {


//  val sparkConf = new SparkConf()
//  sparkConf.setAppName("createGraph").setMaster("local[*]")
//  val sc = new SparkContext(sparkConf)

  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  val spark = SparkSession.builder()
    .appName(this.getClass.getName)
    .master("local[*]")
    .getOrCreate()

  val conf = new MySVDPP.Conf(25,10,1,5,0.001,0.001,0.01,0.0005)

  def predA(g:Graph[(Array[Double], Array[Double], Double, Double),Double],
           mean:Double, u:Long, i:Long) = {
    val user = g.vertices.filter(_._1 == u).collect()(0)._2
    val item = g.vertices.filter(_._1 == i).collect()(0)._2
    mean + user._3 + item._3 +
      item._1.zip(user._2).map(x => x._1 * x._2).reduce(_ + _)
  }

  def pred(g: Graph[(Array[Double], Array[Double], Double, Double), Double],
           mean:Double, u:Long, i:Long): Double = {
    println("pred ",u,i)
    println( g.vertices)
    //    println("filter user ",g.vertices.filter(_._1 == u).collect() )
    println("g is",g.vertices.filter(_._1 == u).collect().foreach(println(_)))
    val user = g.vertices.filter(_._1 == u).collect()(0)._2
    //    print("now user",user)
    val item = g.vertices.filter(_._1 == i).collect()(0)._2
    var ret =  mean + user._3 + item._3 + item._1.zip(user._2).map(x => x._1 * x._2).reduce(_ + _)
    ret = math.max(ret, conf.minVal)
    ret = math.min(ret, conf.maxVal)
    ret

  }

  def main(args: Array[String]): Unit = {

//    val rddSource: RDD[String] = spark.sparkContext.textFile("src/main/test/Douban-dataset/data/edges3.txt")
    val rddSource: RDD[String] = spark.sparkContext.textFile("src/main/test/Douban-dataset/data/ratings_v1_diff_id.csv")

    val value: RDD[Array[String]] = rddSource.map(line => line.split(","))
    val userNum = value.map(x=>x(0)).distinct().count()
    val itNum = value.map(x=>x(1)).distinct().count()
    print("u num "+userNum+" i num "+itNum)




    val EDGES: RDD[Edge[Double]] = value.map(x => Edge(x(0).toLong, x(1).toLong, x(2).toDouble))

//    val EDGES: RDD[Edge[Double]] = EDGE.filter(x=>x.attr!=None).filter(x=>x.dstId!=None).filter(x=>x.srcId!=None)
// mae 1.507
//      val conf = new MySVDPP.Conf(25,100,1,5,0.001,0.001,0.01,0.0005)
// MAE =1.5X
      // MAE =1.584X
//    val conf = new MySVDPP.Conf(25,50,1,5,0.001,0.0005,0.01,0.0005)
      // MAE = 1.6X
//    val conf = new MySVDPP.Conf(50,50,1,5,0.0005,0.0005,0.0025,0.00075)
      //MAE = 1.5X
//    val conf = new MySVDPP.Conf(50,50,1,5,0.0001,0.0001,0.0005,0.00015)
//    val conf = new MySVDPP.Conf(50,50,1,5,0.001,0.001,0.005,0.0015)
      //2.184x
//    val conf = new MySVDPP.Conf(50,50,1,5,0.01,0.01,0.05,0.015)
//    val conf = new MySVDPP.Conf(50,50,1,5,0.1,0.1,0.5,0.15)
//    val updateconf = new MySVDPP.Conf(20,30,1,5,0.00001,0.00001,0.00005,0.000015)
    val updateconf = conf

    var modl: (Graph[(Array[Double], Array[Double], Double, Double), Double], Double) = MySVDPP.run(EDGES, conf)
    val egesset: EdgeRDD[Double] = modl._1.edges
    println(modl._1.vertices.count())
    println("g is ",modl._1.vertices)

      /**
       * 523986010114,47,4.0
       * prediction ==> 1.0
       * 523986010115, 7, 3.5
       * prediction=> 5.0
       */
      //
    print("hello=================>"+pred(modl._1,modl._2,523986010114L,47L))
    print("hello=================>"+pred(modl._1,modl._2,523986010115L,7L))
//    print("hello=================>"+pred( 523986010114L,47L))
//    print("hello=================>"+pred(523986010115L,7L))
//    print("hello=================>"+pred(modl._1,modl._2,5523986010115L,7L))
//    print("hello=================>"+pred(1L,37737L))
//
//    print("hello=================>"+pred(1L,79876L))

//    modl = MySVDPP.OnLinerun(Edge(1L,79876L,1), updateconf)




//    print("hello Again =================>"+pred( 1L,1027L))
//    print("hello Again =================>"+pred(1L,37737L))
//    print("hello=================>"+pred(1L,79876L))




//
//
//    def pred( u:Long, i:Long): Double = {
////      println("pred ",u,i)
////      println( modl._1.vertices)
////      println("filter user ",modl._1.vertices.filter(_._1 == u).collect() )
//        val user = modl._1.vertices.filter(_._1 == u).collect()(0)._2
//      //    print("now user",user)
//        val item = modl._1.vertices.filter(_._1 == i).collect()(0)._2
//        var ret = modl._2 + user._3 + item._3 + item._1.zip(user._2).map(x => x._1 * x._2).reduce(_ + _)
//        ret = math.max(ret, conf.minVal)
//        ret = math.min(ret, conf.maxVal)
//        ret
//
//    }
//def pred(g: Graph[(Array[Double], Array[Double], Double, Double), Double],mean: Double , u:Long, i:Long): Double = {
//  println("pred ",u,i)
//  println( g.vertices)
//  //    println("filter user ",g.vertices.filter(_._1 == u).collect() )
//  println("g is",g.vertices.filter(_._1 == u).collect().foreach(println(_)))
//  val user = g.vertices.filter(_._1 == u).collect()(0)._2
//  //    print("now user",user)
//  val item = g.vertices.filter(_._1 == i).collect()(0)._2
//  var ret =  mean + user._3 + item._3 + item._1.zip(user._2).map(x => x._1 * x._2).reduce(_ + _)
//  ret = math.max(ret, conf.minVal)
//  ret = math.min(ret, conf.maxVal)
//  ret
//
//}


  }

}

/**
 * gamma =0.1 mse =>e63
 * gamma =0.01 mse =>503
 * gama=0.005 mse=>150
 * gama=0.001 mse=>30
 * gama=0.0005 mse=>49
 * gama=0.0001 mse=>104
 */
